package com.apps.sdses.flink141.jar;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;

/**
 * author: codejiwei
 * date: 2023/9/4
 * desc:
 **/
public class JdbcSourceTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(600000);

        // 配置多张表的查询语句
        String[] queries = new String[] {
                String.format("select id, name, description from %s.%s", "mydb", "products"),
                String.format("select id, name from %s.%s", "mydb", "test_add_table")
        };
        RowTypeInfo rowTypeInfo = new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
        RowTypeInfo rowTypeInfo1 = new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);


//        SingleOutputStreamOperator<Row> inputMySQL = env.createInput(JdbcInputFormat.buildJdbcInputFormat()
//                .setDrivername("com.mysql.jdbc.Driver")
//                .setDBUrl(String.format("jdbc:mysql://%s:%d/%s", "localhost", 3306, "mydb"))
//                .setUsername("root")
//                .setPassword("123456")
//                .setQuery(String.format("select id, name, description from %s.%s", "mydb", "products"))
////                .setRowTypeInfo(new RowTypeInfo())
//                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
//                .finish()
//        ).uid("inputMySQL");

        DataStream<Row> dataStream = env.createInput(JdbcInputFormat.buildJdbcInputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl(String.format("jdbc:mysql://%s:%d/%s", "localhost", 3306, "mydb"))
                .setUsername("root")
                .setPassword("123456")
                .setQuery(String.format("select id, name, description from %s.%s", "mydb", "products"))
                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                .finish()
        ).uid("inputMySQL");


        RowTypeInfo rowTypeInfo2 = new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);


//        inputMySQL.print();

        env.execute();
    }
}
